package com.freez.spark.core.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: PersonFlow.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:22
 * Description: 计算某地的客流量
 */
public class PersonFlow {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("personFlow").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
//
        List<Tuple2<String, Integer>> list = Arrays.asList(new Tuple2<String, Integer>("2020-05-01", 9),
                new Tuple2<String, Integer>("2020-05-02",11),
                new Tuple2<String, Integer>("2020-05-05",33),
                new Tuple2<String, Integer>("2020-05-07",22),
                new Tuple2<String, Integer>("2020-05-06", 12),
                new Tuple2<String, Integer>("2020-05-03", 31),
                new Tuple2<String, Integer>("2020-05-04", 29));
/*

        List<Tuple2<Date, Integer>> dateList = new ArrayList<>();
        List<Tuple2<Date, Integer>> nextDayList = new ArrayList<>();
//
        for (int i = 0; i < list.size();i++){
            Tuple2<String, Integer> strTuple = list.get(i);
            Date date = new Date();
            Calendar calendar = Calendar.getInstance();
            try {
                date = new SimpleDateFormat("yyyy-mm-dd").parse(strTuple._1);
                calendar.setTime(date);
                int day = calendar.get(Calendar.DATE);
                calendar.add(Calendar.DATE, +1);
            } catch (ParseException e) {
                e.printStackTrace();
            }
            Tuple2<Date,Integer> dateTuple = new Tuple2<>(date,strTuple._2);
            Tuple2<Date,Integer> nextDateTuple = new Tuple2<>(calendar.getTime(),strTuple._2);
            dateList.add(dateTuple);
            nextDayList.add(nextDateTuple);
        }
*/

        SimpleDateFormat sft = new SimpleDateFormat("yyyy-mm-dd");
        Calendar calendar = Calendar.getInstance();

        JavaPairRDD<String, Integer> strRDD = jsc.parallelizePairs(list);
        JavaPairRDD<Date, Integer> dateRDD = strRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Date, Integer>() {
            @Override
            public Tuple2<Date, Integer> call(Tuple2<String, Integer> s) throws Exception {
                String str = s._1;
                Date date = sft.parse(str);
                return new Tuple2<>(date,s._2);
            }
        });

        JavaPairRDD<Date, Integer> nextDateRDD = dateRDD.mapToPair(new PairFunction<Tuple2<Date, Integer>, Date, Integer>() {
            @Override
            public Tuple2<Date, Integer> call(Tuple2<Date, Integer> v) throws Exception {
                Date date = v._1;
                calendar.setTime(date);
                calendar.add(Calendar.DATE, +1);
                return new Tuple2<>(calendar.getTime(), v._2);
            }
        });
        JavaPairRDD<Date, Tuple2<Integer, Optional<Integer>>> finalRDD = dateRDD.leftOuterJoin(nextDateRDD);
        JavaRDD resultRDD = finalRDD.map(new Function<Tuple2<Date, Tuple2<Integer, Optional<Integer>>>, Tuple2<Date,Integer>>() {
            @Override
            public Tuple2<Date,Integer> call(Tuple2<Date, Tuple2<Integer, Optional<Integer>>> v1) {
                Integer day1num = v1._2._2.orNull();
                if (day1num!= null){
                    return new Tuple2(v1._1,v1._2._1 - day1num);
                }
                return new Tuple2(v1._1,0);
            }
        });
        System.out.println(resultRDD.collect());
        jsc.close();
    }
}